import java.util.concurrent.ArrayBlockingQueue;

/**
 * @Description: 手写一个消息队列
 * @ClassName: CustomMessageQueue
 * @Author: chenyu
 * @Date: 2025/3/31 11:05
 */
public class CustomMessageQueue {

    private int queueSize = 100;

    //队列
    private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(queueSize);

    public static void main(String[] args) {
        CustomMessageQueue customMessageQueue = new CustomMessageQueue();
        Consumer consumer = customMessageQueue.new Consumer();
        consumer.start();

        for (int i = 0; i < 100; i++) {
            Producer producer = customMessageQueue.new Producer();
            producer.start();
        }
    }

    //消费者
    public class Consumer extends Thread {
        @Override
        public void run() {
            while (true) {
                synchronized (queue) {
                    //队列queue有消息
                    if (queue.size() > 0) {
                        Integer i = queue.poll();
                        System.out.println("[Consumer] 从队列queue中获取任务：" + i + "，队列queue当前长度：" + queue.size());
                        //模拟业务处理
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        //唤醒生产者
                        queue.notify();
                    } else { //队列queue空
                        System.out.println("[Consumer] 队列queue当前为空");
                        try {
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notify();
                        }
                    }
                }
            }
        }
    }

    //生产者
    public class Producer extends Thread {
        @Override
        public void run() {
            synchronized (queue) {
                //queue长度>=size，阻塞，等待唤醒，wait()
                //队列长度
                if (queue.size() >= queueSize) {
                    try {
                        queue.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        queue.notify();
                    }
                } else { //queue长度小于size，添加任务
                    queue.add(queue.size()+1);
                    System.out.println("[Producer] 向队列queue中添加任务：" + queue.size() + "，队列queue当前长度：" + queue.size());
                    //模拟业务处理
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //唤醒消费
                    queue.notify();
                }
            }
        }
    }
}
